package org.lds.hotkey.worker.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.lds.hotkey.client.remote.Client;
import org.lds.hotkey.common.autoconfigure.cache.caffeine.CaffeineLocalCacheFactory;
import org.lds.hotkey.common.autoconfigure.cache.LocalCache;
import org.lds.hotkey.common.model.AppInstanceModel;
import org.lds.hotkey.common.model.ReportRuleModel;
import org.lds.hotkey.common.model.RuleModel;
import org.lds.hotkey.common.model.dto.ResultDTO;
import org.lds.hotkey.common.util.MapUtil;
import org.lds.hotkey.console.remote.ConsoleClient;
import org.lds.hotkey.console.remote.model.BatchReceiveHotkeyRequest;
import org.lds.hotkey.console.remote.model.ReceiveHotkeyRequest;
import org.lds.hotkey.console.remote.model.request.BatchCurrHotKeySaveRequest;
import org.lds.hotkey.console.remote.model.request.CurrHotKeySaveRequest;
import org.lds.hotkey.worker.common.model.SlidingWindow;
import org.lds.hotkey.worker.holder.AppInstanceHolder;
import org.lds.hotkey.worker.holder.RuleHolder;
import org.lds.hotkey.worker.remote.model.ReceiveRuleModelRequest;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Slf4j
@Service
public class ReceiveRuleModelService {
    @Autowired
    private ConsoleClient consoleClient;

    private static final LocalCache<String, Boolean> HOT_KEY_CACHE = new CaffeineLocalCacheFactory().createLocalCache(2);
    private static final LocalCache<String, SlidingWindow> LOCAL_CACHE = new CaffeineLocalCacheFactory().createLocalCache(3000);
    private static final ConcurrentHashMap<String, BlockingQueue<ReportRuleModel>> keyReportRuleModelMap = new ConcurrentHashMap<>(256);
    public ResultDTO receiveRuleModel(ReceiveRuleModelRequest request) {
        log.info("接收上报数据 data={}", request);
        ReportRuleModel reportRuleModel = request.getReportRuleModel();
        if (Objects.isNull(HOT_KEY_CACHE.get(reportRuleModel.getKey()))) {
            BlockingQueue<ReportRuleModel> reportRuleModelDeque = keyReportRuleModelMap.computeIfAbsent(reportRuleModel.getKey(), key -> new ArrayBlockingQueue<>(500));
            reportRuleModelDeque.offer(reportRuleModel);
        }
        return ResultDTO.success();
    }

    public void consume() {
        // keyRuleModelMap
        Map<String, RuleModel> keyRuleModelMap = keyRuleModelMap();

        // appIdReceiveHotkeyRequestMap
        Map<Long, List<ReceiveHotkeyRequest>> appIdReceiveHotkeyRequestMap = appIdReceiveHotkeyRequestMap(keyRuleModelMap);

        // 发送
        appIdReceiveHotkeyRequestMap.forEach((appId, receiveHotkeyRequests) -> {
            List<AppInstanceModel> appInstanceModels = AppInstanceHolder.getAppInstanceModels(appId);
            appInstanceModels.forEach(appInstanceModel -> {
                // 客户端感知热key
                Client.receiveHotkey(appInstanceModel.buildAddress(), BatchReceiveHotkeyRequest.builder().requests(receiveHotkeyRequests).build());
                // 控制台感知热key
                consoleClient.saveCurrHotKey(BatchCurrHotKeySaveRequest.builder()
                        .requests(receiveHotkeyRequests.stream().map(receiveHotkeyRequest -> CurrHotKeySaveRequest.builder()
                                .appId(appId)
                                .key(receiveHotkeyRequest.getKey())
                                .duration(receiveHotkeyRequest.getDuration())
                                .build()).collect(Collectors.toList()))
                        .build());
            });
        });
    }

    private Map<String, RuleModel> keyRuleModelMap() {
        Map<String, RuleModel> keyRuleModelMap = new HashMap<>();

        keyReportRuleModelMap.forEach((key, reportRuleModelQueue) -> {
            SlidingWindow slidingWindow = getSlidingWindow(key);
            if (Objects.nonNull(slidingWindow)) {
                List<ReportRuleModel> reportRuleModels = new ArrayList<>();
                reportRuleModelQueue.drainTo(reportRuleModels);

                if (CollectionUtils.isNotEmpty(reportRuleModels) && Objects.isNull(HOT_KEY_CACHE.get(key))) {
                    int sum = sum(reportRuleModels);
                    boolean isHot = slidingWindow.addCount(sum);
                    if (isHot) {
                        HOT_KEY_CACHE.put(key, true);

                        log.info("热key产生 key={}", key);
                        RuleModel ruleModel = RuleHolder.getRuleModel(reportRuleModels.get(0).getRuleId());
                        if (Objects.nonNull(ruleModel)) {
                            keyRuleModelMap.put(key, ruleModel);
                        }
                    }
                }
            }
        });

        return keyRuleModelMap;
    }

    private Map<Long, List<ReceiveHotkeyRequest>> appIdReceiveHotkeyRequestMap(Map<String, RuleModel> keyRuleModelMap) {
        Map<Long, List<ReceiveHotkeyRequest>> appIdReceiveHotkeyRequestMap = new HashMap<>();
        keyRuleModelMap.forEach((key, ruleModel) -> MapUtil.fillListMap(appIdReceiveHotkeyRequestMap, ruleModel.getAppId(),
                ReceiveHotkeyRequest.builder().ruleId(ruleModel.getId()).key(key).duration(ruleModel.getDuration()).build()));
        return appIdReceiveHotkeyRequestMap;
    }

    private int sum(List<ReportRuleModel> reportRuleModels) {
        return reportRuleModels.stream().mapToInt(a -> a.getCount()).sum();
    }

    private SlidingWindow getSlidingWindow(String key) {
        SlidingWindow slidingWindow = LOCAL_CACHE.get(key);
        if (Objects.nonNull(slidingWindow)) return slidingWindow;

        BlockingQueue<ReportRuleModel> reportRuleModels = keyReportRuleModelMap.get(key);
        if (CollectionUtils.isEmpty(reportRuleModels)) return  null;

        ReportRuleModel reportRuleModel = reportRuleModels.peek();
        if (Objects.isNull(reportRuleModel)) return null;

        RuleModel ruleModel = RuleHolder.getRuleModel(reportRuleModel.getRuleId());
        if (Objects.isNull(ruleModel)) return null;

        SlidingWindow newSlidingWindow = new SlidingWindow(ruleModel.getInterval() * 1000, ruleModel.getThreshold());
        LOCAL_CACHE.put(key, newSlidingWindow);
        return newSlidingWindow;
    }
}
